package com.poetic.network.flow;

import com.poetic.network.flow.aggregate.CountAgg;
import com.poetic.network.flow.domain.ApacheLogEvent;
import com.poetic.network.flow.domain.UrlViewCount;
import com.poetic.network.flow.function.WindowResultFunction;
import com.poetic.network.flow.process.TopNHotUrls;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.regex.Pattern;

/**
 * <pre>
 *  实现“热门页面浏览数”的统计，也就是读取服务器日志中的每
 *  一行 log，统计在一段时间内用户访问每一个 url 的次数，然后排序输出显示。
 *
 * 具体做法为：
 *          每隔 5 秒，输出最近 10 分钟内访问量最多的前 N 个 URL。
 * Created by lianghuikun on 2020-09-15.
 * </pre>
 *
 * @author lianghuikun
 */
public class NetworkFlowTask {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设定 Time 类型为 EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // 为了打印到控制台的结果不乱序，我们配置全局的并发为 1，这里改变并发对结果正确性没有影响
        env.setParallelism(1);
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/apache.log";
        env.readTextFile(path)
                .map(new MapFunction<String, ApacheLogEvent>() {
                    @Override
                    public ApacheLogEvent map(String value) throws Exception {
                        String[] data = value.split(" ");
                        // 日期转时间戳 17/05/2015:10:05:03
                        SimpleDateFormat format = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
                        long ts = format.parse(data[3]).getTime();
                        return ApacheLogEvent.builder()
                                .ip(data[0])
                                .userId(data[2])
                                .eventTime(ts)
                                .method(data[5])
                                .url(data[6])
                                .build();
                    }
                })
                // 过滤掉 css和js
                .filter(new FilterFunction<ApacheLogEvent>() {
                    @Override
                    public boolean filter(ApacheLogEvent value) throws Exception {

                        String pattern = "^((?!\\.(css|js)$).)*$";
                        return Pattern.matches(pattern, value.getUrl());
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ApacheLogEvent>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<ApacheLogEvent>() {
                            @Override
                            public long extractTimestamp(ApacheLogEvent element, long recordTimestamp) {
                                return element.getEventTime();
                            }
                        }))
                .keyBy(new KeySelector<ApacheLogEvent, String>() {
                    @Override
                    public String getKey(ApacheLogEvent value) throws Exception {
                        return value.getUrl();
                    }
                })
                .timeWindow(Time.minutes(10), Time.seconds(5))
                .aggregate(new CountAgg(), new WindowResultFunction())
//                .print();
                .keyBy(new KeySelector<UrlViewCount, Long>() {
                    @Override
                    public Long getKey(UrlViewCount value) throws Exception {
                        return value.getWindowEnd();
                    }
                })
                .process(new TopNHotUrls(5))
                .print();

        env.execute("network flow task");
    }
}
